package com.wh.springkafka.admin;

import com.google.gson.Gson;
import com.wh.springkafka.util.PropertyManager;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.acl.AclBindingFilter;
import org.apache.kafka.common.config.ConfigResource;

import java.util.*;
import java.util.concurrent.ExecutionException;

public class AdminSample {

    public static final String TOPIC_NAME = "wanghao-topic";
    public static final Gson gson = new Gson();
    public static final String kafkaServerIp = PropertyManager.getProperty("KafkaZKServerIp");

    public static void main(String[] args) {
        // 创建AdminClient客户端
        AdminClient adminClient = adminClient();
        createTopic(adminClient);
        listTopic(adminClient);
//        delTopic(adminClient);
//        listTopic(adminClient);


//        descTopic(adminClient);
//        descTopicConfig(adminClient);
//        updateTopicConfig(adminClient);
//        descTopicConfig(adminClient);
//        updateTopicConfig2(adminClient);
//        descTopicConfig(adminClient);

//        descTopic(adminClient);
//        addPartitions(adminClient,2);
//        descTopic(adminClient);



    }
    //创建AdminClient客户端
    public static AdminClient adminClient(){
        Properties properties = new Properties();
//        172.17.0.6  111.229.16.81 kafkaHost
        properties.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,kafkaServerIp+":9092");
        AdminClient adminClient = AdminClient.create(properties);
        return adminClient;
    }

    // 创建topic实例
    public static void createTopic(AdminClient adminClient){
        // 副本因子
        Short s = 1;
        NewTopic newTopic = new NewTopic(TOPIC_NAME,2,s);
        CreateTopicsResult result = adminClient.createTopics(Arrays.asList(newTopic));
        System.out.println("CreateTopicsResult：" +result);
    }
    // 查询topic列表
    public static void listTopic(AdminClient adminClient)  {
        try {
            // 普通查看
            Set<String> names = adminClient.listTopics().names().get();
            System.out.println("names："+names);

            // 过滤查看 Internal
            ListTopicsOptions options = new ListTopicsOptions();
            options.listInternal(true);
            Set<String> names2 = adminClient.listTopics(options).names().get();
            System.out.println("names{options}："+names2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }

    // 删除topic实例
    public static void delTopic(AdminClient adminClient){
        DeleteTopicsResult result = adminClient.deleteTopics(Arrays.asList(TOPIC_NAME));
        try {
            System.out.println("DeleteTopicsResult：" +result.all().get());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

    }
    // 获取topic实例
    public static void descTopic(AdminClient adminClient){

        DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Arrays.asList(TOPIC_NAME));
        try {
            System.out.println(gson.toJson(describeTopicsResult.all().get()));
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

    }
    // 获取topic 配置信息
    public static void descTopicConfig(AdminClient adminClient){
        ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC,TOPIC_NAME);
        DescribeConfigsResult describeConfigsResult = adminClient.describeConfigs(Arrays.asList(resource));
        try {
            System.out.println(gson.toJson(describeConfigsResult.all().get()));
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

    }
    // 修改topic 配置信息
    public static void updateTopicConfig(AdminClient adminClient){
        Map<ConfigResource,Config> map = new HashMap<>();

        ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC,TOPIC_NAME);
        ConfigEntry entry = new ConfigEntry("preallocate","true");

        Config config = new Config(Arrays.asList(entry));
        map.put(resource,config);
        AlterConfigsResult alterConfigsResult = adminClient.alterConfigs(map);
        try {
            System.out.println(alterConfigsResult.all().get());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }
    // 修改topic 配置信息
    public static void updateTopicConfig2(AdminClient adminClient){
        Map<ConfigResource,Collection<AlterConfigOp>> map = new HashMap<>();

        ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC,TOPIC_NAME);
        ConfigEntry entry = new ConfigEntry("preallocate","false");

        Config config = new Config(Arrays.asList(entry));
        AlterConfigOp op = new AlterConfigOp(entry,AlterConfigOp.OpType.SET);
        map.put(resource,Arrays.asList(op));
        AlterConfigsResult alterConfigsResult = adminClient.incrementalAlterConfigs(map);
        try {
            System.out.println(alterConfigsResult.all().get());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }
    // 添加 partitions
    public static void addPartitions(AdminClient adminClient,int numPartitions){
        NewPartitions partitions = NewPartitions.increaseTo(numPartitions);
        Map<String, NewPartitions> partitionsMap = new HashMap<>();
        partitionsMap.put(TOPIC_NAME,partitions);
        adminClient.createPartitions(partitionsMap);
    }
}
